package transform;

import bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.Test;

public class Flink_Transform_reduce {

    @Test
    public void test1() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> input = env.readTextFile("input/sensor.txt");
        SingleOutputStreamOperator<SensorReading> map = input.map((MapFunction<String, SensorReading>) value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2]));
        });
        KeyedStream<SensorReading, String> keyedStream = map.keyBy(SensorReading::getId);
        SingleOutputStreamOperator<SensorReading> reduce = keyedStream.reduce((ReduceFunction<SensorReading>) (value1, value2) -> {
            if (value1.getTemp() > value2.getTemp()) {
                return value1;
            }
            return value2;
        });
        reduce.print();
        env.execute();
    }
}
